package com.atguigu.day04;

import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.connector.file.table.FileSystemTableSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.time.Duration;
import java.time.ZoneId;

public class Flink06_Sink_File {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.从端口读数据
        DataStreamSource<String> streamSource = env.socketTextStream("localhost", 9999);
//        DataStreamSource<String> streamSource = env.fromElements("aaa", "bbb", "ccc");


        //TODO 3.将数据写入文件
        FileSink<String> fileSink = FileSink.forRowFormat(new Path("output"), new SimpleStringEncoder<String>())
                //指定生成文件的前缀和后缀
                .withOutputFileConfig(new OutputFileConfig("atguigu-",".log"))
                //指定生成目录的周期
                .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd-HH", ZoneId.of("UTC+8")))
                //文件滚动策略:  1分钟 或 1m
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(Duration.ofSeconds(1))
                                .withMaxPartSize(new MemorySize(1024*1024))
                                .build()
                )
                .build();

        streamSource.sinkTo(fileSink);

        env.execute();
    }
}
